///数据中台，用于数据的转发,本质上数据流是一个队列

#include "streamstation.hpp"

using namespace systemos;

stream::stream(const char* cName, StreamDirectTPDF tStreamDirect, uint16_t usSize)
{
	Name = cName;
	Size = usSize;
	mQueue = new Queue(1, usSize);
	StreamDirect = tStreamDirect;
	streamstation::Register(this);
}

stream::~stream(void)
{
	streamstation::Unregister(this);
	delete mQueue;
}

void stream::Public(void *tag)
{
	mQueue->Send(tag);
}

bool stream::Subscribe(void *tag)
{
	return(mQueue->Recevie(tag));
}

static stream* mStream[STREAMSTATION_MAX_SURPPORT] = {0};
///>转发相应的数据流订阅对象
static void StreamPublic(stream *tag, uint8_t ucExcept, uint8_t *ucMem)
{
	uint8_t i;
	for(i = 0; i < STREAMSTATION_MAX_SURPPORT; i++)
	{
		if(!mStream[i])
		{
			continue;
		}
		if(i == ucExcept)
		{
			continue;
		}
		if(memcmp(mStream[i]->Name, tag->Name, strlen(tag->Name)) == 0 && mStream[i]->StreamDirect == stream::STREAM_DIRECT_SUBSCRIBE)
		{
			mStream[i]->Public(ucMem);
		}
	}
}

static void Thread(void *args)
{
	uint8_t i;
	static uint8_t mem[STREAMSTATION_MAX_BUFFLEN];
	
	while(1)
	{
		Method::Sleep(1);
		for(i = 0; i < STREAMSTATION_MAX_SURPPORT; i++)
		{
			if(!mStream[i])
			{
				continue;
			}
			if(mStream[i]->StreamDirect == stream::STREAM_DIRECT_SUBSCRIBE)
			{
				continue;
			}
			if(mStream[i]->Subscribe(mem))///对外发布的数据进行转发
			{
				StreamPublic(mStream[i], i, mem);
			}
		}
	}
}

void streamstation::Init(void)
{
	threados *t1 = new threados(Thread);
	t1->Name = "stream";
	t1->Prio = 0;
	t1->StackDeep = 1024;
	t1->Start();
}

bool streamstation::Register(stream *tStream)///每一个数据流都要注册
{
	uint8_t i;
	for(i = 0; i < STREAMSTATION_MAX_SURPPORT; i++)
	{
		if(!mStream[i])
		{
			mStream[i] = tStream;
			return(true);
		}
	}
	return(false);
}

void streamstation::Unregister(stream *tStream)
{
	uint8_t i;
	for(i = 0; i < STREAMSTATION_MAX_SURPPORT; i++)
	{
		if(mStream[i] == tStream)
		{
			mStream[i] = 0;
			break;
		}
	}
}
